package com.pie4cloud.pie.job.biz.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.pie4cloud.pie.job.api.dto.PlayBillVo;
import com.pie4cloud.pie.job.api.entity.*;
import com.pie4cloud.pie.job.biz.gather.service.ConfigDecodeService;
import com.pie4cloud.pie.job.biz.gather.service.ConfigFtpService;
import com.pie4cloud.pie.jpa.BaseDao;
import com.pie4cloud.pie.jpa.BaseService;
import com.pie4cloud.pie.jpa.page.PageBean;
import com.pie4cloud.pie.jpa.page.PageForm;
import com.pie4cloud.pie.jpa.specification.SimpleSpecificationBuilder;
import com.pie4cloud.pie.bus.api.entity.DataKafConsumer;
import com.pie4cloud.pie.bus.api.feign.RemoteKafkaManageService;
import com.pie4cloud.pie.common.core.constant.SecurityConstants;
import com.pie4cloud.pie.common.core.util.R;
import com.pie4cloud.pie.job.biz.dao.DispenseCongfigDao;
import com.pie4cloud.pie.job.biz.dao.DispenseHistoryDao;
import com.pie4cloud.pie.job.biz.dao.DispenseTimingDao;
import com.pie4cloud.pie.job.biz.service.DispenseCongfigService;
import com.pie4cloud.pie.job.biz.service.JobInfoService;
import com.pie4cloud.pie.job.biz.service.KafkaJobRelationService;
import com.pie4cloud.pie.job.biz.thread.ScheduleThread;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.annotation.CacheEvict;
import org.springframework.data.domain.Sort;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.stereotype.Service;

import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
@RequiredArgsConstructor
public class DispenseCongfigServiceImpl extends BaseService<DispenseCongfigEntity,Long> implements DispenseCongfigService {
	@Autowired
	private DispenseCongfigDao dispenseCongfigDao;
	@Autowired
	private DispenseHistoryDao dispenseHistoryDao;
	@Autowired
	private DispenseTimingDao dispenseTimingDao;
	@Autowired
	private ScheduleThread scheduleThread;
	private final ConfigFtpService configFtpService;
	@Autowired
	private JobInfoService jobInfoService;
	private final RemoteKafkaManageService remoteKafkaManageService;
	@Autowired
	private KafkaJobRelationService kafkaJobRelationService;
	@Autowired
	private ConfigDecodeService configDecodeService;

	@Override
	public BaseDao<DispenseCongfigEntity,Long> getBaseDao() {
		return dispenseCongfigDao;
	}

	@Override
	public DispenseCongfigEntity getById(Long id) {
		return super.getById(id);
	}

	@Override
	public List<DispenseCongfigEntity> findAll() {
		return dispenseCongfigDao.findAll();
	}

	public PageBean<DispenseCongfigEntity> page(Page page, DispenseCongfigEntity dispenseCongfigEntity) {
		SimpleSpecificationBuilder specificationBuilder = new SimpleSpecificationBuilder();
		Specification specification = specificationBuilder.generateSpecification();
		Sort sort = Sort.by(Sort.Direction.DESC, "createTime");
		PageForm pageForm = new PageForm((int) page.getCurrent(), (int) page.getSize());
		PageBean pageBean = this.getPage(specification, pageForm, sort);
		List<DispenseCongfigEntity> list=pageBean.getRecords();
		for(int i=0;i<list.size();i++){
			if(null==list.get(i).getDispenseTiming()){
				list.get(i).setDispenseTiming(new DispenseTimingEntity());
			}
			if(null==list.get(i).getDispenseHistory()){
				list.get(i).setDispenseHistory(new DispenseHistoryEntity());
			}
		}
		pageBean.setRecords(list);
		return pageBean;
	}

	public DispenseCongfigEntity save(DispenseCongfigEntity dispenseCongfigEntity) {
		if(dispenseCongfigEntity.getIsHistory()==0){
			if(null!=dispenseCongfigEntity.getDispenseHistory()&&null!=dispenseCongfigEntity.getDispenseHistory().getId()){
				dispenseHistoryDao.deleteById(dispenseCongfigEntity.getDispenseHistory().getId());
			}
			dispenseCongfigEntity.setDispenseHistory(null);
		}
		if(dispenseCongfigEntity.getIsRealTime()==0){
			if(null!=dispenseCongfigEntity.getDispenseTiming()&&null!=dispenseCongfigEntity.getDispenseTiming().getId()){
				dispenseTimingDao.deleteById(dispenseCongfigEntity.getDispenseHistory().getId());
			}
			dispenseCongfigEntity.setDispenseTiming(null);
		}
		dispenseCongfigEntity = dispenseCongfigDao.save(dispenseCongfigEntity);
		if(dispenseCongfigEntity.getIsHistory()==2){
			JobInfoEntity jobInfoEntity=new JobInfoEntity();
			jobInfoEntity.setType("DISPENSE");
			jobInfoEntity.setExecutorHandler("dispenseHistoryHandler");
			ConfigFtp configFtp=configFtpService.getById(dispenseCongfigEntity.getFtpId());
			dispenseCongfigEntity.setConfigFtp(configFtp);
			jobInfoEntity.setExecutorParam(JSON.toJSONString(dispenseCongfigEntity));
			scheduleThread.handT(jobInfoEntity);
		}else {
			if(0==dispenseCongfigEntity.getIsRealTime()){
				DataKafConsumer dataKafConsumer=new DataKafConsumer();
				dataKafConsumer.setWorkNum(4);
				dataKafConsumer.setTopic("GATHER_"+dispenseCongfigEntity.getPlayBillId());
				dataKafConsumer.setKafGroup("DISPENSE_"+dispenseCongfigEntity.getId());
				dataKafConsumer.setHandler("dispenseRealTimeKafkaHandler");
				dataKafConsumer.setServiceName("pie-job-executor");
				dataKafConsumer.setTriggerStatus(1);
				R<DataKafConsumer> dataKafConsumerR= remoteKafkaManageService.save(dataKafConsumer, SecurityConstants.FROM_IN);
				if(0==dataKafConsumerR.getCode()){
					KafkaJobRelation kafkaJobRelation=new KafkaJobRelation();
					kafkaJobRelation.setConsId(dataKafConsumerR.getData().getConsId());
					kafkaJobRelation.setId("DISPENSE_"+dispenseCongfigEntity.getId());
					kafkaJobRelationService.save(kafkaJobRelation);
				}
			}else {
				jobInfoService.start(dispenseCongfigEntity.getDispenseTiming());
			}
		}
		return dispenseCongfigEntity;
	}

	@Override
	@CacheEvict(value = "DISPENSE", key = "'id='+#dispenseCongfigEntity.dispenseTiming.id",beforeInvocation = true)
	public DispenseCongfigEntity update(DispenseCongfigEntity dispenseCongfigEntity) {
		if(dispenseCongfigEntity.getIsHistory()==0){
			dispenseCongfigEntity.setDispenseHistory(null);
		}
		if(dispenseCongfigEntity.getIsRealTime()==0){
			dispenseCongfigEntity.setDispenseTiming(null);
		}
		dispenseCongfigEntity = dispenseCongfigDao.save(dispenseCongfigEntity);
		if(dispenseCongfigEntity.getIsHistory()==2){
			JobInfoEntity jobInfoEntity=new JobInfoEntity();
			jobInfoEntity.setType("DISPENSE");
			jobInfoEntity.setExecutorHandler("dispenseHistoryHandler");
			ConfigFtp configFtp=configFtpService.getById(dispenseCongfigEntity.getFtpId());
			dispenseCongfigEntity.setConfigFtp(configFtp);
			jobInfoEntity.setExecutorParam(JSON.toJSONString(dispenseCongfigEntity));
			scheduleThread.pushRedis(jobInfoEntity);
		}else {
			if(0==dispenseCongfigEntity.getIsRealTime()){
				KafkaJobRelation kafkaJobRelation=kafkaJobRelationService.getById("DISPENSE_"+dispenseCongfigEntity.getId());
				DataKafConsumer dataKafConsumer=new DataKafConsumer();
				dataKafConsumer.setWorkNum(4);
				dataKafConsumer.setTopic("GATHER_"+dispenseCongfigEntity.getPlayBillId());
				dataKafConsumer.setKafGroup("DISPENSE_"+dispenseCongfigEntity.getId());
				dataKafConsumer.setHandler("dispenseRealTimeKafkaHandler");
				dataKafConsumer.setServiceName("pie-job-executor");
				dataKafConsumer.setTriggerStatus(1);
				if(null!=kafkaJobRelation){
					dataKafConsumer.setConsId(kafkaJobRelation.getConsId());
				}
				R<DataKafConsumer> dataKafConsumerR= remoteKafkaManageService.updateById(dataKafConsumer, SecurityConstants.FROM_IN);
				if(null==kafkaJobRelation&&0==dataKafConsumerR.getCode()){
					kafkaJobRelation=new KafkaJobRelation();
					kafkaJobRelation.setConsId(dataKafConsumerR.getData().getConsId());
					kafkaJobRelation.setId("DISPENSE_"+dispenseCongfigEntity.getId());
					kafkaJobRelationService.save(kafkaJobRelation);
				}
			}else {
				jobInfoService.start(dispenseCongfigEntity.getDispenseTiming());
			}
		}
		return dispenseCongfigEntity;
	}

	public Boolean removeById(Long id) {

		DispenseCongfigEntity dispenseCongfigEntity = dispenseCongfigDao.findById(id).get();
		if(1==dispenseCongfigEntity.getIsRealTime()){
			jobInfoService.stop(dispenseCongfigEntity.getDispenseTiming().getId());
		}else {
			KafkaJobRelation kafkaJobRelation=kafkaJobRelationService.getById("DISPENSE_"+dispenseCongfigEntity.getId());
			R<DataKafConsumer> dataKafConsumerR= remoteKafkaManageService.removeById(kafkaJobRelation.getConsId(),SecurityConstants.FROM_IN);
			if(0==dataKafConsumerR.getCode()){
				kafkaJobRelationService.delById(kafkaJobRelation.getId());
			}
		}
		dispenseCongfigDao.deleteById(id);
		return true;
	}

	public PlayBillVo findByPlayBillId(Long playBillId){
		List<DispenseCongfigEntity> dispenses=dispenseCongfigDao.findDispenseCongfigEntitiesByPlayBillId(playBillId);
		List<ConfigDecode> decodes=configDecodeService.findConfigDecodesByPlayBillId(playBillId);
		PlayBillVo playBillVo=new PlayBillVo();
		playBillVo.setDispenses(dispenses);
		playBillVo.setDecodes(decodes);
		return playBillVo;
	}
}
